/*
 * Copyright (C) 2012  Romain DUBOIS
 * 
 * This file is part of NeMoS - Network and Modular Software
 * 
 * NeMoS is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * NeMoS is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with NeMoS.  If not, see <http://www.gnu.org/licenses/>.
 */
package nemos.remote.rsa.message.impl;

import static nemos.remote.rsa.message.MessageConfigurationConstants.FRAMEWORK_EVENT_TOPIC_ROOT;
import static nemos.remote.rsa.message.MessageConfigurationConstants.SERVER;
import static nemos.remote.rsa.message.MessageConfigurationConstants.TRANSPORT_ID_PROPERTY;

import java.lang.reflect.InvocationTargetException;
import java.rmi.RemoteException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import nemos.remote.IInterceptor;
import nemos.remote.RemoteCall;
import nemos.remote.rsa.internal.RsaMessage;
import nemos.remote.rsa.message.data.EndpointMessage;
import nemos.remote.rsa.message.data.FrameworkMessage;
import nemos.remote.rsa.message.data.InitMessage;
import nemos.remote.rsa.message.data.InvocationMessage;
import nemos.remote.rsa.message.data.Message;
import nemos.remote.rsa.message.data.ResponseMessage;
import nemos.remote.rsa.message.data.ShutdownMessage;
import nemos.remote.rsa.message.transport.ITransport;
import nemos.remote.util.EplEndpointProvider;
import nemos.remote.util.RemoteUtil;
import nemos.util.log.NemosLogger;
import nemos.util.osgi.FilterBuilder;

import org.osgi.framework.BundleContext;
import org.osgi.framework.Constants;
import org.osgi.framework.FrameworkUtil;
import org.osgi.framework.ServiceReference;
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;
import org.osgi.util.tracker.ServiceTracker;

/**
 * MessageProcessor
 * 
 */
public class MessageProcessor {

    /** The endpoint notifier */
    EplEndpointProvider mEndpointNotifier;

    /** The interceptors tracker */
    private ServiceTracker<IInterceptor, IInterceptor> mInterceptors;
    /** The event admin tracker */
    private ServiceTracker<EventAdmin, EventAdmin> mEventAdmin;

    /** The transport */
    private final ITransport mTransport;
    /** The connected frameworks UUID */
    private final Set<String> mConnectedFrameworks;

    /** The bundle context */
    private BundleContext mContext;
    /** The logger */
    private final NemosLogger mLogger;

    /** The executor service */
    private final ExecutorService mExecutor = Executors.newCachedThreadPool();

    /**
     * Constructor.
     * 
     * @param pTransport
     *            The transport
     */
    public MessageProcessor(final ITransport pTransport) {
        super();

        mTransport = pTransport;
        mConnectedFrameworks = new HashSet<String>();
        mLogger = new NemosLogger(MessageProcessor.class);
        mEndpointNotifier = new EplEndpointProvider();
    }

    /**
     * Start the processor.
     * 
     * @param pContext
     *            The bundle context
     */
    public void start(final BundleContext pContext) {
        mContext = pContext;

        mEndpointNotifier.activate(pContext);
        mInterceptors = new ServiceTracker<IInterceptor, IInterceptor>(pContext,
                IInterceptor.class, null);
        mInterceptors.open();
        mEventAdmin = new ServiceTracker<EventAdmin, EventAdmin>(pContext, EventAdmin.class, null);
        mEventAdmin.open();
    }

    /**
     * A framework joined or left the server.
     * 
     * @param pMessage
     *            The framework message
     */
    protected void frameworkEvent(final FrameworkMessage pMessage) {
        // Update data
        boolean lNotify = false;
        if (pMessage.isAdded()) {
            lNotify = mConnectedFrameworks.add(pMessage.getFrameworkUUID());
        } else if (pMessage.isRemoved()) {
            lNotify = mConnectedFrameworks.remove(pMessage.getFrameworkUUID());
        }

        // Notify if needed and possible
        final EventAdmin lEventAdmin = mEventAdmin.getService();
        if (lNotify && lEventAdmin != null) {
            final String lTopic = FRAMEWORK_EVENT_TOPIC_ROOT + pMessage.getAction().name();
            final Map<String, Object> lProperties = new HashMap<String, Object>();
            lProperties.put(SERVER, Boolean.FALSE);
            lProperties.put(TRANSPORT_ID_PROPERTY, mTransport.getId());
            lProperties.put(Constants.FRAMEWORK_UUID, pMessage.getFrameworkUUID());

            lEventAdmin.postEvent(new Event(lTopic, lProperties));
        }
    }

    /**
     * Process a local service call from a remote framework.
     * 
     * @param pMessage
     *            The call message
     */
    protected void callService(final InvocationMessage pMessage) {
        Message lMessageToSend = null;
        try {

            // Collect target service reference
            final String lFilter = FilterBuilder.getStringFilter(Constants.SERVICE_ID,
                    String.valueOf(pMessage.getServiceId()));
            final ServiceReference<?>[] lRefs = mContext.getAllServiceReferences((String) null,
                    lFilter);

            // Unknown service (may have been unregistered during transport
            if (lRefs == null || lRefs.length != 1) {
                final String lText = mLogger.getTranslator().getMessage(RsaMessage.UNKNOWN_SERVICE,
                        Long.valueOf(pMessage.getId()), Long.valueOf(pMessage.getServiceId()));
                throw new RemoteException(lText);
            }

            // Prepare call data
            final RemoteCall lCall = new RemoteCall(pMessage.getCallingFrameworkUuid());
            final ServiceReference<?> lRef = lRefs[0];
            lCall.setReference(lRef);
            lCall.setMethod(pMessage.getMethod(lRef.getBundle()));
            lCall.setParameters(pMessage.getArgs(lRef.getBundle()));
            lCall.setService(mContext.getService(lRef));

            // Perform call
            try {
                performCall(lCall);

                lMessageToSend = new ResponseMessage(pMessage.getId(), lCall.getResult(), false);
            } catch (final InvocationTargetException e) {
                lMessageToSend = new ResponseMessage(pMessage.getId(), e.getCause(), true);
            } finally {
                mContext.ungetService(lRef);
            }

        } catch (final RemoteException e) {
            lMessageToSend = new ResponseMessage(pMessage.getId(), e);
        } catch (final Exception e) {
            final Long lMessageID = Long.valueOf(pMessage.getId());
            mLogger.error(RsaMessage.INVOCATION_ERROR, e, lMessageID);
            final String lText = mLogger.getTranslator().getMessage(RsaMessage.INVOCATION_ERROR,
                    lMessageID);
            lMessageToSend = new ResponseMessage(pMessage.getId(), new RemoteException(lText, e));
        }

        // Try to send response
        try {
            mTransport.send(lMessageToSend);
        } catch (final RemoteException e) {
            mLogger.error(RsaMessage.UNABLE_TO_SEND_RESPONSE, e, Long.valueOf(pMessage.getId()));
        }
    }

    /**
     * Perform the service call with interceptors management.
     * 
     * @param pCall
     *            The remote call to perform
     * @throws InvocationTargetException
     *             If an invocation error occurred
     * @throws Exception
     *             If an abnormal error occurred
     */
    private void performCall(final RemoteCall pCall) throws InvocationTargetException, Exception {

        boolean lPerformCall = true;

        // Call all interceptors in their natural order
        final Iterator<Entry<ServiceReference<IInterceptor>, IInterceptor>> lIt = mInterceptors
                .getTracked().entrySet().iterator();
        while (lIt.hasNext()) {
            final Entry<ServiceReference<IInterceptor>, IInterceptor> lEntry = lIt.next();
            final ServiceReference<?> lRef = lEntry.getKey();

            // Check if service invocation has to be done
            final Object lRanking = lRef.getProperty(Constants.SERVICE_RANKING);
            if (lPerformCall && lRanking != null && ((Integer) lRanking).intValue() < 0) {
                pCall.setResult(pCall.getMethod().invoke(pCall.getService(), pCall.getParameters()));
                lPerformCall = false;
            }

            // Call only interested interceptors
            final String lFilter = (String) lRef.getProperty(IInterceptor.INTERCEPTOR_SCOPE);
            if (lFilter == null || FrameworkUtil.createFilter(lFilter).match(pCall.getReference())) {
                lEntry.getValue().interceptInvocation(pCall);
            }
        }

        // Ensure to perform the call
        if (lPerformCall) {
            pCall.setResult(pCall.getMethod().invoke(pCall.getService(), pCall.getParameters()));
        }
    }

    /**
     * Process an incoming message.
     * 
     * @param pMessage
     *            The message
     * @return If it should be the last process call
     */
    public boolean process(final Message pMessage) {
        boolean lStop = false;

        // Incoming remote call
        if (pMessage instanceof InvocationMessage) {
            mExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    callService((InvocationMessage) pMessage);
                }
            });
        }
        // Remote endpoint added or removed
        else if (pMessage instanceof EndpointMessage) {
            final EndpointMessage lMsg = (EndpointMessage) pMessage;
            mExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    mEndpointNotifier.notifyListeners(lMsg.getEndpoint(), lMsg.isAdded());
                }
            });
        }
        // Remote framework joining or leaving
        else if (pMessage instanceof FrameworkMessage) {
            final FrameworkMessage lMsg = (FrameworkMessage) pMessage;
            lStop = RemoteUtil.getFrameworkUuid().equals(lMsg.getFrameworkUUID())
                    && lMsg.isRemoved();
            mExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    frameworkEvent(lMsg);
                }
            });
        }
        // Init server data
        else if (pMessage instanceof InitMessage) {
            final InitMessage lMsg = ((InitMessage) pMessage);
            mTransport.setLabel(lMsg.getTransportLabel());
            mConnectedFrameworks.clear();
            mConnectedFrameworks.addAll(lMsg.getConnectedFrameworks());
        }
        // Servers will shutdown
        else if (pMessage instanceof ShutdownMessage) {
            mTransport.close();
        }

        // Unexpected message type
        else {
            mLogger.error(RsaMessage.UNEXPECTED_MESSAGE_TYPE, pMessage.getClass().getName());
        }

        return lStop;
    }

    /**
     * Stop the processor.
     */
    public void stop() {
        // Stop executions
        mExecutor.shutdown();

        // Stop helpers
        mInterceptors.close();
        mEventAdmin.close();
        mEndpointNotifier.deactivate();
    }

    /**
     * Get the connected frameworks.
     * 
     * @return The connected frameworks
     */
    public Set<String> getConnectedFrameworks() {
        return mConnectedFrameworks;
    }
}
